package com.example.springbatch.service;

import com.example.springbatch.common.KafkaReader;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.batch.item.adapter.ItemReaderAdapter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.PartitionOffset;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import java.text.MessageFormat;
import java.util.List;

/**
 * Created by Liaopan on 2018/9/10.
 */
@Service
@Slf4j
public class KafkaService {

    @Autowired
    @Qualifier("kafkaTemplate")
    private KafkaTemplate template;

    @Autowired
    KafkaReader reader;

    public void sendToKafka(final String data) {
        final ProducerRecord<String, String> record = new ProducerRecord<>("tt", "1", data);

        ListenableFuture<SendResult<Integer, String>> future = template.send(record);
        future.addCallback(new ListenableFutureCallback<SendResult<Integer, String>>() {

            @Override
            public void onSuccess(SendResult<Integer, String> result) {
                System.out.println(result);
                System.out.println(data);
            }

            @Override
            public void onFailure(Throwable ex) {
                System.out.println(ex);
                System.out.println(record);

                System.out.println(data);
            }

        });
    }

    @KafkaListener(id = "test", topicPartitions =
            {@TopicPartition(topic = "tt", partitions = "1",
                    partitionOffsets = @PartitionOffset(partition = "0", initialOffset = "0"))
            },containerFactory = "kafkaListenerContainerFactory")
    public void listen(List<?> record) {
        System.out.println("SimpleConsumer收到消息：" + record);
        reader.setRecord(record);
    }

}